package com.feng.rabbitmq.four;

import com.feng.rabbitmq.utiils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

import static java.lang.System.currentTimeMillis;
import static java.lang.System.setOut;

/**
 * 发布确认模式
 * 使用的时间比较哪种确认方式是最好的
 * 1、单个确认
 * 2、批量确认
 * 3、异步批量确认
 */
public class ConfirmMessage {

    //批量发送消息的数量
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
    //    1、单个确认
        //发布了 1000 条数据，总共花费了 336ms
//        ConfirmMessage.publishMessageIndividually();
    //    2、批量确认
        //发布了 1000 条数据，总共花费了 50ms
//        ConfirmMessage.publishMessageBatch();
    //    3、异步批量确认
        //发布了 1000 条数据，总共花费了 40ms（因为异步，所以打印这个可能在中间）
        ConfirmMessage.publishMessageAsync();
    }
    /**
     *单个确认
     * 发布了 1000 条数据，总共花费了 336ms
     */
    public static void publishMessageIndividually() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        //队列名
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布时间
        channel.confirmSelect();
        //开始时间
        long begin = currentTimeMillis();

        //批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息发送就马上确认
            boolean flag = channel.waitForConfirms();
            if (flag){
                System.out.println("消息发送成功");
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布了 " + MESSAGE_COUNT + " 条数据，" + "总共花费了 " + (end - begin) + "ms");;
    }

    /**
     * 批量发布确认
     * 发布了 1000 条数据，总共花费了 50ms
     */
    public static void publishMessageBatch() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        //队列名
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布时间
        channel.confirmSelect();
        //开始时间
        long begin = currentTimeMillis();

        //批量发送消息
        //批量确认消息大小
        int batchSize = MESSAGE_COUNT/10;
        for (int i = 1; i <= MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());
            //批量大小达到batchSize的时候批量确认一次
            if (i%batchSize==0){
                boolean flag = channel.waitForConfirms();
                if (flag){
                    System.out.println("消息发送成功");
                }
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布了 " + MESSAGE_COUNT + " 条数据，" + "总共花费了 " + (end - begin) + "ms");;
    }

    /**
     * 异步发布确认
     * 发布了 1000 条数据，总共花费了 40ms
     * 发布了 1000 条数据，总共花费了 37ms
     */
    public static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMQUtils.getChannel();
        //队列名
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布时间
        channel.confirmSelect();

        /**
         * 线程安全有序的一个哈希表适用于高并发的情况下
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目只要给到序号
         * 3.支持高并发（多线程)
         */
        ConcurrentSkipListMap<Long,String> listMap = new ConcurrentSkipListMap();
        //消息确认成功，回调函数
        ConfirmCallback ackCallBack = (deliveryTag,multiple) -> {
            if (multiple){
                ConcurrentNavigableMap<Long, String> confirmed = listMap.headMap(deliveryTag);
                confirmed.clear();
            }else {
                listMap.remove(deliveryTag);
            }
            System.out.println("确认的消息："+deliveryTag);;
        };
        //消息确认失败，回调函数
        ConfirmCallback nackCallBack = (deliveryTag,multiple) -> {
            String message = listMap.get(deliveryTag);
            System.out.println("未确认的消息："+message+"\n未确认的消息标记是："+deliveryTag);;
        };
        //准备消息的监听器，监听哪些消息成功了，哪些失败了
        channel.addConfirmListener(ackCallBack,nackCallBack);

        //开始时间
        long begin = currentTimeMillis();

        //批量发送消息
        for (int i = 1; i <= MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("",queueName,null,message.getBytes());
            //此处记录下所有要发送的消息消息的总和
            listMap.put(channel.getNextPublishSeqNo(),message);
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布了 " + MESSAGE_COUNT + " 条数据，" + "总共花费了 " + (end - begin) + "ms");;
    }

}
